BackTrader 海龟交易策略
介绍
海龟交易策略是一种经典的交易系统。
本文策略是在《Backtrader来啦：常见案例汇总》一文中提出的，我进行了复现。
本文只用于数学、编程研究，不提供交易指导。
《Backtrader来啦：常见案例汇总》一文中总结道，策略细节为：
指标计算：20日最高、最低、收盘价计算真实波动 ATR
计算20日最高、最低加，构建唐奇安通道
交易信号：
- 入场：价格突破20日最高价
- 加仓：价格继续上涨至 0.5 倍 ATR ，再次加仓，加仓次数不超过 3 次；
- 止损：价格回落 2 倍 ATR 时止损离场；
- 止盈：价格突破 10 日最低点时止盈离场；
其中：海龟交易策略之所以称之为交易系统，因为包括了加仓、止盈止损的逻辑在内。
代码实现
具体代码实现参照《Backtrader来啦：常见案例汇总》一文，融入我的量化系统，具体代码如下：
import backtrader as bt
from newstock.data.mongo.mongo_data_manager import MongoDataManager
from newstock.date.stock_date import StockDate
from newstock.market.Exchange import SZSEExchange
from newstock.market.symbol import Symbol
import pandas as pd
"""
海龟交易系统
"""
class TestStrategy(bt.Strategy):
params = dict(
N1=20, # 唐奇安通道上轨的t
N2=10, # 唐奇安通道下轨的t
printlog=False,
)
def log(self, txt, dt=None, doprint=False):
"""Logging function for this strategy"""
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print("%s, %s" % (dt.isoformat(), txt))
def __init__(self):
# Keep a reference to the "close" line in the data[0] dataseries
self.dataclose = self.datas[0].close
# To keep track of pending orders
self.order = None
self.buy_count = 0 # 记录买入次数
self.last_price = 0 # 记录买入价格
self.close = self.datas[0].close
self.high = self.datas[0].high
self.low = self.datas[0].low
# 计算唐奇安通道上轨：过去20日的最高价
self.DonchianH = bt.ind.Highest(self.high(-1), period=self.p.N1, subplot=True)
# 计算唐奇安通道下轨：过去10日的最低价
self.DonchianL = bt.ind.Lowest(self.low(-1), period=self.p.N2, subplot=True)
# 生成唐奇安通道上轨突破：close>DonchianH，取值为1.0；反之为 -1.0
self.CrossoverH = bt.ind.CrossOver(self.close(0), self.DonchianH, subplot=False)
# 生成唐奇安通道下轨突破:
self.CrossoverL = bt.ind.CrossOver(self.close(0), self.DonchianL, subplot=False)
# 计算 ATR
self.TR = bt.ind.Max(
(self.high(0) - self.low(0)), # 当日最高价-当日最低价
abs(self.high(0) - self.close(-1)), # abs(当日最高价−前一日收盘价)
abs(self.low(0) - self.close(-1)), # abs(当日最低价-前一日收盘价)
)
self.ATR = bt.ind.SimpleMovingAverage(self.TR, period=self.p.N1, subplot=False)
bt.indicators.MACDHisto(self.datas[0])
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
# Buy/Sell order submitted/accepted to/by broker - Nothing to do
return
# 交易完成
# Attention: broker could reject order if not enough cash
if order.status in [order.Completed]:
if order.isbuy():
self.log(
"BUY EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f"
% (order.executed.price, order.executed.value, order.executed.comm)
)
self.buyprice = order.executed.price # 买入价格
self.buycomm = order.executed.comm # 买入手续费
elif order.issell():
self.log(
"SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f"
% (order.executed.price, order.executed.value, order.executed.comm)
)
self.bar_executed = len(self) # 买入日期
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log("Order Canceled/Margin/Rejected")
# Write down: no pending order
self.order = None
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log("OPERATION PROFIT, GROSS %.2f, NET %.2f" % (trade.pnl, trade.pnlcomm))
def next(self):
# Simply log the closing price of the series from the reference
self.log("Close, %.2f" % self.dataclose[0])
# Check if an order is pending ... if yes, we cannot send a 2nd one
if self.order:
return
# 计算加仓单位，写死10手
buy_count = 100
# 入场：价格突破上轨线且空仓时，做多
if self.position.size == 0:
if self.CrossoverH > 0 and self.buy_count == 0:
self.order = self.buy(size=buy_count)
self.last_price = self.position.price # 记录买入价格
self.buy_count = 1 # 记录本次交易价格
elif self.position.size > 0:
# 多单加仓:价格上涨了买入价的0.5的ATR且加仓次数少于等于3次
if (
self.datas[0].close > self.last_price + 0.5 * self.ATR[0]
and self.buy_count <= 3
):
self.order = self.buy(size=buy_count)
self.last_price = self.position.price # 获取买入价格
self.buy_count = self.buy_count + 1
# 多单止损：当价格回落2倍ATR时止损平仓
elif self.datas[0].close < (self.last_price - 2 * self.ATR[0]):
self.order = self.sell(size=self.position.size)
self.buy_count = 0
# 多单止盈：当价格突破10日最低点时止盈离场 平仓
elif self.CrossoverL < 0:
self.order = self.sell(size=self.position.size)
self.buy_count = 0
def stop(self):
self.log(
"Ending Value %.2f" % (self.broker.getvalue()),
doprint=True,
)
if __name__ == "__main__":
cerebro = bt.Cerebro()
print("Starting Portfolio Value: %.2f" % cerebro.broker.getvalue())
mongoManager = MongoDataManager()
df = mongoManager.getStockPeriodFromDB(
Symbol(SZSEExchange, "000001"),
StockDate.today().previousDays(500),
StockDate.today(),
)
df["date"] = pd.to_datetime(df["trade_date"], format="%Y%m%d")
data = bt.feeds.PandasData(dataname=df, datetime="date") # type: ignore
df.dropna()
# 0.1% ... divide by 100 to remove the %
cerebro.broker.setcommission(commission=0.001)
# Python 3.10 修复 module 'collections' has no attribute 'Iterable' 开始
import collections
collections.Iterable = collections.abc.Iterable
# Python 3.10 修复 module 'collections' has no attribute 'Iterable' 完成
# 策略参数优化
# cerebro.optstrategy(TestStrategy, maperiod=range(10, 31))
# 策略运行
cerebro.addstrategy(TestStrategy)
cerebro.adddata(data)
# Add a FixedSize sizer according to the stake
cerebro.addsizer(bt.sizers.FixedSize, stake=10)
cerebro.run()
cerebro.plot(style="bar", volume=False)
print("Final Portfolio Value: %.2f" % cerebro.broker.getvalue())
其中，在每次买入量中我写死了 size=10，应该是 10 手。
效果
我发现效果并不太好，每次都买在最高点上。止损看起来还行。
代码改进
参照《量化投资2：基于backtrader实现完整海龟法则量化回测》，原来在海龟交易法中，每次加仓的单位是与波动情况挂钩的。引入这部分实现。之后的策略变为：
class TestSizer(bt.Sizer):
params = (("stake", 1),)
def _getsizing(self, comminfo, cash, data, isbuy):
if isbuy:
return self.p.stake
position = self.broker.getposition(data)
if not position.size:
return 0
else:
return position.size
return self.p.stake
class TestStrategy(bt.Strategy):
params = dict(
N1=20, # 唐奇安通道上轨的t
N2=10, # 唐奇安通道下轨的t
printlog=False,
)
def log(self, txt, dt=None, doprint=False):
"""Logging function for this strategy"""
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print("%s, %s" % (dt.isoformat(), txt))
def __init__(self):
# Keep a reference to the "close" line in the data[0] dataseries
self.dataclose = self.datas[0].close
# To keep track of pending orders
self.order = None
self.buy_count = 0 # 记录买入次数
self.last_price = 0 # 记录买入价格
self.newstake = 0
self.close = self.datas[0].close
self.high = self.datas[0].high
self.low = self.datas[0].low
# 计算唐奇安通道上轨：过去20日的最高价
self.DonchianH = bt.ind.Highest(self.high(-1), period=self.p.N1, subplot=False)
# 计算唐奇安通道下轨：过去10日的最低价
self.DonchianL = bt.ind.Lowest(self.low(-1), period=self.p.N2, subplot=False)
# 生成唐奇安通道上轨突破：close>DonchianH，取值为1.0；反之为 -1.0
self.CrossoverH = bt.ind.CrossOver(self.close(0), self.DonchianH, subplot=False)
# 生成唐奇安通道下轨突破:
self.CrossoverL = bt.ind.CrossOver(self.close(0), self.DonchianL, subplot=False)
# 计算 ATR
self.TR = bt.ind.Max(
(self.high(0) - self.low(0)), # 当日最高价-当日最低价
abs(self.high(0) - self.close(-1)), # abs(当日最高价−前一日收盘价)
abs(self.low(0) - self.close(-1)), # abs(当日最低价-前一日收盘价)
)
self.ATR = bt.ind.SimpleMovingAverage(self.TR, period=self.p.N1, subplot=False)
bt.indicators.MACDHisto(self.datas[0])
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
# Buy/Sell order submitted/accepted to/by broker - Nothing to do
return
# 交易完成
# Attention: broker could reject order if not enough cash
if order.status in [order.Completed]:
if order.isbuy():
self.log(
"BUY EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f"
% (order.executed.price, order.executed.value, order.executed.comm)
)
self.buyprice = order.executed.price # 买入价格
self.buycomm = order.executed.comm # 买入手续费
elif order.issell():
self.log(
"SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f"
% (order.executed.price, order.executed.value, order.executed.comm)
)
self.bar_executed = len(self) # 买入日期
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log("Order Canceled/Margin/Rejected")
# Write down: no pending order
self.order = None
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log("OPERATION PROFIT, GROSS %.2f, NET %.2f" % (trade.pnl, trade.pnlcomm))
def next(self):
# Simply log the closing price of the series from the reference
self.log("Close, %.2f" % self.dataclose[0])
# Check if an order is pending ... if yes, we cannot send a 2nd one
if self.order:
return
# 入场：价格突破上轨线且空仓时，做多
if self.CrossoverH > 0 and self.buy_count == 0:
self.newstake = self.broker.getvalue() * 0.01 / self.ATR
self.newstake = int(self.newstake / 100) * 100
self.sizer.p.stake = self.newstake
self.buy_count = 1
self.order = self.buy()
# 加仓:价格上涨了买入价的0.5的ATR且加仓次数少于等于3次
elif (
self.datas[0].close > self.last_price + 0.5 * self.ATR[0]
and self.buy_count > 0
and self.buy_count < 4
):
self.newstake = self.broker.getvalue() * 0.01 / self.ATR
self.newstake = int(self.newstake / 100) * 100
self.sizer.p.stake = self.newstake
self.buy_count = self.buy_count + 1
self.order = self.buy()
# 出场
elif self.CrossoverL < 0 and self.buy_count > 0:
self.order = self.sell(size=int(self.position.size))
self.buy_count = 0
# 止损
elif (
self.datas[0].close < (self.last_price - 2 * self.ATR[0])
and self.buytime > 0
):
self.order = self.sell(size=int(self.position.size))
self.buy_count = 0
def stop(self):
self.log(
"Ending Value %.2f" % (self.broker.getvalue()),
doprint=True,
)
效果
再次运行，效果如下：
从中可以看到，交易时机点不变，变化的是仓位根据波动性而变。整体上最终效果要比之前好一些。
ATR 周期改为 14
在《【手把手教你】用backtrader量化回测海龟交易策略》一文中，将 ATR 的周期改为了 14，我的执行完后，整体效果变化不明显，图就不贴了。
感悟
开仓条件：入场：价格突破20日最高价，这是一个价格突破策略，但是只适合于价格呈上升趋势的，在图中价格下跌或水平趋势是，总是容易开仓在局部高点。
更换开仓条件
灵感，更改开仓条件，变为均线金叉买入，其它不变。
其中：
- 交易变得特别频繁，其实效果也没比双均线好多少
- 海龟策略的止损比双均线止损要更加灵敏一些